package com.shanxi.cz.module.biz.framework.mqtt.config;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;

@Slf4j
@Configuration
public class MqttSubscribeConfig {

    @Autowired
    private MqttPahoClientFactory mqttClientFactory;

    @Value("${mqtt.client-id}")
    private String clientId;

    @Value("${mqtt.topic}")
    private String topic;

    // 消息输入通道
    @Bean
    public MessageChannel mqttInputChannel() {
        DirectChannel channel = new DirectChannel();
        channel.setBeanName("mqttInputChannel");
        return channel;
    }

    // MQTT消息订阅适配器
    @Bean
    public MqttPahoMessageDrivenChannelAdapter mqttInbound() {
        String[] topics = {topic};
        // 为每个主题设置QoS
        int[] qos = new int[topics.length];
        for (int i = 0; i < topics.length; i++) {
            qos[i] = 0;
        }
        MqttPahoMessageDrivenChannelAdapter adapter =
                new MqttPahoMessageDrivenChannelAdapter(clientId,
                        mqttClientFactory, topics);

        adapter.setCompletionTimeout(30000);
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setQos(0);
        adapter.setOutputChannel(mqttInputChannel());
        adapter.setErrorChannelName("errorChannel");

        log.info("MQTT订阅器配置完成 - 客户端ID: {}", clientId);
        log.info("MQTT订阅配置完成 - 主题: {}", String.join(", ", topics));
        return adapter;
    }
}
